package com.example.demo07cache.component;

import cn.hutool.core.util.StrUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.function.Consumer;

/**
 * @author wangxun
 */
@Component
public class RedisTemplateExt {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private RedisLock redisLock;

    public byte[] get(final String key) {
        return redisTemplate.execute((RedisCallback<byte[]>) con -> con.get(key.getBytes()));
    }

    public Long bitCount(final String key) {
        return redisTemplate.execute((RedisCallback<Long>) con -> con.bitCount(key.getBytes()));
    }

    /**
     * 计算 key 对应的 value 的二进制位中，bit = 1 的个数
     *
     * @param key       RedisKey
     * @param startByte byte 的起始偏移量
     * @param endByte   byte 的最终偏移量
     * @return key 对应的 value 的二进制位中，byte 偏移量 [startByte, endByte] 之间 bit = 1 的个数
     */
    public Long bitCount(String key, long startByte, long endByte) {
        if (startByte > endByte) {
            return 0L;
        }
        return redisTemplate.execute((RedisCallback<Long>) con -> con.bitCount(key.getBytes(), startByte, endByte));
    }

    private static final long THREE = 3;
    private static final long SEVEN = 7;
    private static final long EIGHT = 8;

    /**
     * 计算 key 对应的 value 的二进制位中，bit = 1 的个数，这个操作是非原子的
     *
     * @param key            RedisKey
     * @param startBitOffset bit 的起始偏移量
     * @param endBitOffset   bit 的最终偏移量
     * @return key 对应的 value 的二进制位中，bit 偏移量 [startBitOffset, endBitOffset] 之间 bit = 1 的个数
     */
    public Long bitCountUnAtomic(String key, long startBitOffset, long endBitOffset) {
        if (startBitOffset < 0 || endBitOffset < 1 || startBitOffset >= endBitOffset) {
            throw new IllegalArgumentException("startOffset 不能大于等于 endOffset");
        }

        long[] start = computeQuotientAndRemainder(startBitOffset);
        long[] end = computeQuotientAndRemainder(endBitOffset);

        if (start[0] == end[0]) {
            if (start[1] == 0 && end[1] == 7) {
                return bitCount(key, start[0], start[0]);
            } else {
                return redisLock.bitCount(key, startBitOffset, endBitOffset);
            }
        } else {
            if (start[1] == 0 && end[1] < 7) {
                return bitCount(key, start[0], end[0] - 1) + redisLock.bitCount(key, end[0] * 8, endBitOffset);
            } else if (start[1] > 0 && end[1] == 7) {
                return bitCount(key, start[0] + 1, end[0]) + redisLock.bitCount(key, startBitOffset, start[0] * 8 + 7);
            } else if (start[1] == 0 && end[1] == 7) {
                return bitCount(key, start[0], end[0]);
            } else {
                if (start[0] + 1 == end[0]) {
                    return redisLock.bitCount(key, startBitOffset, start[0] * 8 + 7) + redisLock.bitCount(key, end[0] * 8, endBitOffset);
                } else {
                    return redisLock.bitCount(key, startBitOffset, start[0] * 8 + 7) + bitCount(key, start[0] + 1, end[0] - 1) + redisLock.bitCount(key, end[0] * 8, endBitOffset);
                }
            }
        }
    }


    /**
     * 计算 number / 8 的商和余数
     *
     * @param number 被除数
     * @return 商和余数
     */
    private long[] computeQuotientAndRemainder(long number) {
        // 商
        long quotient = number >> THREE;
        // 余数
        long remainder = number & SEVEN;
        return new long[]{quotient, remainder};
    }

    public Long bitOp(RedisStringCommands.BitOperation op, String saveKey, String... desKey) {
        byte[][] bytes = new byte[desKey.length][];
        for (int i = 0; i < desKey.length; i++) {
            bytes[i] = desKey[i].getBytes();
        }
        return redisTemplate.execute((RedisCallback<Long>) con -> con.bitOp(op, saveKey.getBytes(), bytes));
    }

    /**
     * 增量迭代 key
     * SCAN cursor [MATCH pattern] [COUNT count] [TYPE type]
     *
     * @param pattern 例如: a* 或者 *a*
     * @param count   每次遍历的个数
     * @note Lettuce 帮我们实现了循环 scan 命令(不需要我们自己循环 scan)，正因为如此，如果 pattern 匹配的 key 接近于无穷，返回的集合中的元素接近于无穷，可能因为集合中元素太多导致 OOM
     */
    public void scan(String pattern, int count, Consumer<byte[]> action) {
        redisTemplate.execute((RedisCallback<Boolean>) con -> {
            Cursor<byte[]> cursor = con.scan(new ScanOptions.ScanOptionsBuilder().match(pattern).count(count).build());
            cursor.forEachRemaining(action);
            return Boolean.TRUE;
        });
    }

    /**
     * 增量迭代 Set 的元素
     * SSCAN cursor [MATCH pattern] [COUNT count] [TYPE type]
     *
     * @param key      Redis Key
     * @param pattern  例如: a* 或者 *a*
     * @param count    每次遍历的个数
     * @note Lettuce 帮我们实现了循环 sscan 命令(不需要我们自己循环 sscan)，正因为如此，如果 pattern 匹配的 key 接近于无穷，返回的集合中的元素接近于无穷，可能因为集合中元素太多导致 OOM
     */
    public void sScan(String key, String pattern, int count, Consumer<byte[]> action) {
        redisTemplate.execute((RedisCallback<Boolean>) con -> {
            Cursor<byte[]> cursor = con.sScan(key.getBytes(StandardCharsets.UTF_8), new ScanOptions.ScanOptionsBuilder().match(pattern).count(count).build());
            cursor.forEachRemaining(action);
            return Boolean.TRUE;
        });
    }

    /**
     * 增量迭代 Set 的元素
     * HSCAN cursor [MATCH pattern] [COUNT count] [TYPE type]
     *
     * @param key      Redis Key
     * @param pattern  例如: a* 或者 *a*
     * @param count    每次遍历的个数
     * @note Lettuce 帮我们实现了循环 sscan 命令(不需要我们自己循环 sscan)，正因为如此，如果 pattern 匹配的 key 接近于无穷，返回的集合中的元素接近于无穷，可能因为集合中元素太多导致 OOM
     */
    public void hScan(String key, String pattern, int count, Consumer<Map.Entry<byte[], byte[]>> action) {
        redisTemplate.execute((RedisCallback<Boolean>) con -> {
            Cursor<Map.Entry<byte[], byte[]>> cursor = con.hScan(key.getBytes(StandardCharsets.UTF_8), new ScanOptions.ScanOptionsBuilder().match(pattern).count(count).build());
            cursor.forEachRemaining(action);
            return Boolean.TRUE;
        });
    }

    /**
     * 增量迭代 Set 的元素
     * HSCAN cursor [MATCH pattern] [COUNT count] [TYPE type]
     *
     * @param key      Redis Key
     * @param pattern  例如: a* 或者 *a*
     * @param count    每次遍历的个数
     * @note Lettuce 帮我们实现了循环 sscan 命令(不需要我们自己循环 sscan)，正因为如此，如果 pattern 匹配的 key 接近于无穷，返回的集合中的元素接近于无穷，可能因为集合中元素太多导致 OOM
     */
    public void zScan(String key, String pattern, int count, Consumer<RedisZSetCommands.Tuple> action) {
        redisTemplate.execute((RedisCallback<Boolean>) con -> {
            Cursor<RedisZSetCommands.Tuple> cursor = con.zScan(key.getBytes(StandardCharsets.UTF_8), new ScanOptions.ScanOptionsBuilder().match(pattern).count(count).build());
            cursor.forEachRemaining(action);
            return Boolean.TRUE;
        });
    }

    public boolean scriptLoad(String script) throws Exception {
        if (StrUtil.isBlank(script.trim())) {
            throw new Exception("脚本不能为空白");
        }
        Boolean result = redisTemplate.execute((RedisCallback<Boolean>) con -> {
            con.scriptLoad(script.trim().getBytes(StandardCharsets.UTF_8));
            return Boolean.TRUE;
        });
        return Objects.isNull(result) ? false : result;
    }

    public boolean scriptExists(String... scriptShas) throws Exception {
        for (String script : scriptShas) {
            if (StrUtil.isBlank(script.trim())) {
                throw new Exception("脚本不能为空白");
            }
        }
        Boolean result = redisTemplate.execute((RedisCallback<Boolean>) con -> {
            con.scriptExists(scriptShas);
            return Boolean.TRUE;
        });
        return Objects.isNull(result) ? false : result;
    }

    public boolean scriptFlush() {
        Boolean result = redisTemplate.execute((RedisCallback<Boolean>) con -> {
            con.scriptFlush();
            return Boolean.TRUE;
        });
        return Objects.isNull(result) ? false : result;
    }
}
